package com.test.study.ToolsUtils.strom.workdsum;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FileUtils;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * 本类用于监听某一个目录下文件的变化
 * 
 * @author huyong
 *
 */
public class DataSourceSpout extends BaseRichSpout {
	private static final long serialVersionUID = 1L;
	@SuppressWarnings({ "unused", "rawtypes" })
	private Map conf;
	@SuppressWarnings("unused")
	private TopologyContext context;
	private SpoutOutputCollector collector;

	@SuppressWarnings("rawtypes")
	@Override
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		this.conf = conf;
		this.context = context;
		this.collector = collector;
	}

	@Override
	public void nextTuple() {
		// 这里监听指定文件夹的文件的变化 获取指定目录下 已txt结尾的文件 true 表示是否递归获取
		Collection<File> listFiles = FileUtils.listFiles(new File("/Users/huyong/App/file/testDir"),
				new String[] { "txt" }, true);
		if (listFiles != null && listFiles.size() > 0) {
			for (File file : listFiles) {
				try {
					List<String> readLines = FileUtils.readLines(file, Charset.defaultCharset());
					if (readLines != null && readLines.size() > 0) {
						int num=0;
						for (String line : readLines) {
							// 将每一行 发射出去
							this.collector.emit(new Values(line),num++);
							// 这里读完之后后面在次读的时候会出现重复 解决办法是把文件名的后缀改了
						}
					}
					FileUtils.moveFile(file, new File(file.getAbsolutePath() + System.currentTimeMillis()));
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("line"));
	}

}
